package bsp.diff;

import java.io.IOException;
import java.util.StringTokenizer;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hama.bsp.BSPPeer;
import org.apache.hama.bsp.sync.SyncException;

import util.*;

public class DiffStepOne<T0 extends Parsable<T0>, T1 extends Parsable<T1>, T2 extends Parsable<T2>>
		extends DiffMapper<T0, T1, T2, DiffMessage<T1>> {

	private T0 cur;
	protected int flg;

	//private T0 pre;

	public DiffStepOne(DiffComputation<T0, T1, T2> di) {
		super(di);

		flg = 0;
		chunkey = -1;

	}

	// copy constructor
	public DiffStepOne(DiffStepOne<T0, T1, T2> other) {
		super(other.diff);
		flg = other.flg;
		chunkey = other.chunkey;
	}

	// public void transferStates(
	// DiffMapper<T0, T1, T2> other )
	// throws IOException, SyncException, InterruptedException{
	// other.otimsAcc = this.otimsAcc;
	// other.otimsAccPrew = this.otimsAccPrew;
	// other.chunkey =this.chunkey;
	// }

	@Override
	public void map(BSPPeer<LongWritable, Text, Text, NullWritable, Text> peer)
			throws IOException, SyncException, InterruptedException {

		LongWritable key = new LongWritable(-1);

		Text line = new Text();
		while (peer.readNext(key, line)) {
			if (chunkey < 0) {
				chunkey = key.get();
				peer.getConfiguration().setLong("chunkey", chunkey);
			}
			StringTokenizer itr = new StringTokenizer(line.toString().trim(),
					TOKEN);

			while (itr.hasMoreTokens()) {

				String tmp = itr.nextToken();
				T0 ins = p.getInstanceOfT0();
				cur = ins.parse(tmp);

				if (flg == 0) { // first chunk , 1st step
					if (chunkey == 0) {
						otimsAcc = cons_c;//otimes.evaluate(cons_c, q.evaluate(cur));
					}
					flg++;
				}
				otimsAccPrew = otimsAcc;
				otimsAcc = otimes.evaluate(otimsAcc, q.evaluate(cur));
				//pre = cur;

			}
		}

		if (chunkey >= 0) {
			// System.out.println("+++++DEBUG++++++ ");

			System.out.println("otimsAcc: " + otimsAcc);

			for (int destPeerId = 0; destPeerId < this.numPeers; destPeerId++) {
				// All to all
				addMesg(peer.getPeerName(destPeerId), new Text(
						new DiffMessage<T1>(chunkey, otimsAcc).toString()));
				System.out.println("send msg: "
						+ new DiffMessage<T1>(chunkey, otimsAcc).toString());
			}

		}
	}
}